#include "kc031service.h"

#ifdef USE_MQTT

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include "mosquitto.h"
#include "kc031service.h"
#include "mqtt.h"
#include "aexudp.h"
#include "mod_gpio.h"
#include "sgw_device.h"

// 定义运行标志决定是否需要结束
static int running = 1;
static char g_sub_topic[200]="";
static char g_pub_topic[200]="";
static int mqtt_connected=0;

static char MQTT_ID[200]="kc031";

int get_mqtt_connected()
{
	return mqtt_connected;
}

char *getconf_mqtt_topic()
{
	psus_config *psudata = get_psus_data();

	if(psudata->server_mqtttopic == NULL){
		return "/kc031srv";
	}else{
		return psudata->server_mqtttopic;
	}
}

char *mqtt_get_sub_topic(unsigned char modid)
{
	psus_config *psudata = get_psus_data();
	char *modstr = (modid != aexgpio_get_mod_addr())?get_mod_by_id(get_slot_mod(modid)):psudata->main_devicetype;

	memset(g_sub_topic,0,sizeof(g_sub_topic));
	snprintf(g_sub_topic,sizeof(g_sub_topic),"%1$s/%2$s/%2$s-%3$s/sub",getconf_string(psudata->server_mqtttopic),
		modstr,getconf_string(psudata->main_uuid));
	return g_sub_topic;
}

char *mqtt_get_pub_topic(unsigned char modid)
{
	psus_config *psudata = get_psus_data();
	char *modstr = (modid != aexgpio_get_mod_addr())?get_mod_by_id(get_slot_mod(modid)):psudata->main_devicetype;
	
	memset(g_pub_topic,0,sizeof(g_pub_topic));
	snprintf(g_pub_topic,sizeof(g_pub_topic),"%1$s/%2$s/%2$s-%3$s/pub",getconf_string(psudata->server_mqtttopic),
		modstr,getconf_string(psudata->main_uuid));
	return g_pub_topic;
}

void mqtt_build_topic()
{
	psus_config *psudata = get_psus_data();

	memset(MQTT_ID,0,sizeof(MQTT_ID));
	snprintf(MQTT_ID,sizeof(MQTT_ID)-1,"%s-%s",psudata->main_devicetype,getconf_string(psudata->main_uuid));
}

// char *mqtt_get_recive_topic()
// {
// 	return g_sub_topic;
// }

// char *mqtt_get_send_topic()
// {
// 	return g_pub_topic;
// }

void mqtt_connect_callback(struct mosquitto *mosq, void *obj, int rc)
{
	psus_config *psudata = get_psus_data();

	psus_hint("MQTT: on_connect rc=%d",rc);
	if(!rc){
		char buf[150];
		// 订阅主题
		// 参数：句柄、id、订阅的主题、qos
		//连接成功后，需要订阅相应的主题
		
		if(mosquitto_subscribe(mosq, NULL, mqtt_get_sub_topic(aexgpio_get_mod_addr()), 2)){
			psus_error("Set the topic error!");
		}else{
			psus_hint("SUBSCRIBE:%s",mqtt_get_sub_topic(aexgpio_get_mod_addr()));
		}
		mqtt_connected = 1;

		memset(buf,0,sizeof(buf));
		snprintf(buf,sizeof(buf)-1,"%1$s/%2$s/%2$s-%3$s/sub",getconf_string(psudata->server_mqtttopic),
			psudata->Heating_topic,getconf_string(psudata->main_uuid));
		if(mosquitto_subscribe(mosq, NULL, buf, 2)){
			psus_error("Set the heating topic error!");
		}else{
			psus_hint("SUBSCRIBE:%s",buf);
		}
		unsigned char modid=0;
		for(modid = SGW_TYPE_MIN;modid < SGW_TYPE_MAX;modid++){
			if(mosquitto_subscribe(mosq, NULL, mqtt_get_sub_topic(modid), 2)){
				psus_error("Set the topic error!\n");
			}else{
				psus_hint("SUBSCRIBE:%s",mqtt_get_sub_topic(modid));
			}
		}
		aexgpio_set_online_status(1);
		aexudp_buid_heart_packnet();
	}
}

void mqtt_disconnect_callback(struct mosquitto *mosq, void *obj, int rc)
{
	
	psus_hint("MQTT: my_disconnect_callback rc=%d",rc);
	running = 0;
	mqtt_connected = 0;
	aexgpio_set_online_status(0);
}

void mqtt_subscribe_callback(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
{
	// psus_hint("MQTT: on_subscribe");
	if(granted_qos != NULL){
		int i;
		psus_hint("subscribed (mid: %d): %d", mid, granted_qos[0]);
		for(i=1; i < qos_count; i++){
			// psus_hint(", %d", granted_qos[i]);
		}
		// psus_hint("");
	}
}

void mqtt_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{
	char cmd[255];
	int  cmdlen=sizeof(cmd);
	psus_config *pusdata = get_psus_data();

	//psus_hint("Call the function: on_message");
	psus_hint("MQTT %s : %s", (char *)msg->topic, (char *)msg->payload);

	if(0 == strcmp(msg->payload, "quit")){
		mosquitto_disconnect(mosq);
	}
	//下面根据mqtt消息解析执行相应的动作

	memset(cmd,0,sizeof(cmd));
	HexDecode(msg->payload,strlen(msg->payload),cmd,&cmdlen);
	pusdata->hcsdata.last_cmd_from = AEX_LCF_MQTT;		//设置命令的最后来源为 MQTT
	PPKTPROPO pktpropo = aexudp_get_pktpropo(cmd);
	if(pktpropo != NULL){
		AEXUDP_DATA_SET_SRC(cmd,0);		//从 MQTT 来的命令修改来源为 0x00，方便解析处理
		pktpropo->func(cmd,AEXUDP_GET_PACKET_LEN(cmd));
	}
}

void mqtt_publish_callback(struct mosquitto *mosq, void *obj, int mid)
{
	// psus_hint("Call the function: my_publish_callback");
}

static	struct mosquitto *mosq=NULL;

char *get_pdata_json(char *mbuf,int mlen,unsigned char cmd,unsigned char *pdata,int plen)
{
	char dobuf[30],dibuf[30],aibuf[50],aobuf[50];

	memset(mbuf,0,sizeof(mlen));
	switch (cmd)
	{
	case AEX_CMD_HEART:
	case AEX_CMD_IOSTATUS:
	case AEX_CMD_DOSET:
	case AEX_CMD_DOINVERT:
		{
			int index = 1;
			int do_n = pdata[index++];	//DO的数量
			unsigned char *do_b = &pdata[index];		//DO的缓冲区起始地址
			// psus_hint("HEART:DO%d=%d",do_n,index);
			index += bits_to_bytes(do_n);				//跳过DO的缓冲区
			int di_n = pdata[index++];					//DI的数量
			unsigned char *di_b = &pdata[index];		//DI的缓冲区起始地址
			// psus_hint("HEART:DI%d=%d",di_n,index);
			index += bits_to_bytes(di_n);				//跳过 DI的缓冲区
			int ao_n = pdata[index++];					//AO的数量
			unsigned char *ao_b = pdata[index];		//AO的缓冲区起始地址
			// psus_hint("HEART:AO%d=%d",ao_n,index);
			index += ao_n*AEXGPIO_MAX_AIO_BYTE;								//跳过 AO的缓冲区
			int ai_n = pdata[index++];					//AI的数量
			unsigned char *ai_b = &pdata[index];		//AI的缓冲区起始地址
			// psus_hint("HEART:AI%d=%d",ai_n,index);
			index += ai_n*AEXGPIO_MAX_AIO_BYTE;								//跳过 AI的缓冲区
			unsigned char *gitnb = &pdata[index];		//GIT版本号的起始地址
			// psus_hint("HEART:Git%d:%02X%02X%02X%02X,%s",index,gitnb[0],gitnb[1],gitnb[2],gitnb[3],gitnb+4);
			unsigned long gitn = (gitnb[0]<<24) | (gitnb[1]<<16) | (gitnb[2]<<8) | (gitnb[3]);

			if(plen < gitnb - pdata){
				psus_error("MQTT: Heartbeat data length error!");
				return "";
			}
			sprintf(mbuf,"{\"DO\":\"[%s]\",\"DI\":\"[%s]\",\"AO\":\"[%s]\",\"AI\":\"[%s]\",\"gitn\":\"%u\",\"gitv\":\"%s\"}"
				,get_pdata_bit(dobuf,sizeof(dobuf),do_b,do_n)
				,get_pdata_bit(dibuf,sizeof(dibuf),di_b,di_n)
				,uint32_array_to_csv(aobuf,sizeof(aobuf),ao_b,ao_n)
				,uint32_array_to_csv(aibuf,sizeof(aibuf),ai_b,ai_n)
				,gitn
				,gitnb+4
			);

		}
		break;
	case  AEX_CMD_TRANSDATA:
	{
		int index = 0;
		int uart_n = pdata[index++];	

		sprintf(mbuf,"{\"uart%d\":\"%s\"}",uart_n
			,get_utf8_printable(&pdata[index],plen - 1)					//	
		);

	}
	break;
	default:
		sprintf(mbuf,"{}");
		break;
	}
	return mbuf;
}

void  mqtt_pub(char *buf,int len)
{
	psus_config *pusdata = get_psus_data();
	unsigned char *data = AEXUDP_GET_DATA(buf);
	unsigned char plen = AEXUDP_GET_LEN(buf)-AEXUDP_DATA_GET_PAYLOADINDEX();
	unsigned char *pdata = AEXUDP_DATA_GET_PAYLOAD(data);
	unsigned char dst_addr = AEXUDP_DATA_GET_DST(data);		//接受命令的设备地址，0表示广播命令
	unsigned char src_addr = AEXUDP_DATA_GET_SRC(data);

	char mqttcmd[1024];
	char dphex[512];
	char mbuf[512];
	int dlen=sizeof(dphex);

	memset(mbuf,0,sizeof(mbuf));
	memset(dphex,0, dlen);
	HexEncode(AEXUDP_DATA_GET_PAYLOAD(AEXUDP_GET_DATA(buf)),AEXUDP_GET_LEN(buf)-AEXUDP_DATA_GET_PAYLOADINDEX(),dphex,&dlen);
	memset(mqttcmd,0, sizeof(mqttcmd));
	snprintf(mqttcmd,sizeof(mqttcmd)-1,"{\"cmd\":%d,\"index\":%d,\"len\":%d,\"dst\":%d,\"src\":%d,\"from\":%d,\"msg\":\"%s\",\"pdata\":%s}\r\n",
		(unsigned char)AEXUDP_GET_CMD(buf),
		(unsigned char)AEXUDP_GET_INDEX(buf),
		(unsigned char)AEXUDP_GET_LEN(buf),
		dst_addr,
		src_addr,
		pusdata->hcsdata.last_cmd_from,
		dphex,
		get_pdata_json(mbuf,sizeof(mbuf),AEXUDP_GET_CMD(buf),pdata,plen)
	);
	/*发布消息*/
	psus_hint("PUB:%s:%s",mqtt_get_pub_topic(src_addr),mqttcmd);
	mosquitto_publish(mosq,NULL,mqtt_get_pub_topic(src_addr),strlen(mqttcmd)+1,mqttcmd,0,0);
	// if(1){
	// 	dlen=sizeof(dphex);
	// 	memset(dphex,0, dlen);
	// 	HexEncode(buf,AEXUDP_GET_PACKET_LEN(buf),dphex,&dlen);
	// 	psus_hint("AEXUDP:%s\r",dphex);
	// }
}

void  mqtt_pub_with_topic(char *topic,char *buf,int len)
{
	/*发布消息*/
	psus_hint("PUB:%s:%s",topic,buf);
	mosquitto_publish(mosq,NULL,topic,len,buf,0,0);
}

int mqtt_start()
{
	psus_config *psudata = get_psus_data();
	int ret;

	aexgpio_set_online_status(0);
	mqtt_build_topic();
	// 初始化mosquitto库
	ret = mosquitto_lib_init();
	if(ret){
		psus_hint("MQTT Init lib error!");
		mqtt_connected = 0;
		return -1;
	}

	// 创建一个订阅端实例
	// 参数：id（不需要则为NULL）、clean_start、用户数据
	//struct mosquitto *mosquitto_new( const char * id, bool clean_session, void * obj )
	mosq = mosquitto_new(MQTT_ID, true, NULL);
	if(mosq == NULL){
		psus_hint("MQTT New sub_test error!");
		mosquitto_lib_cleanup();
		mqtt_connected = 0;
		return -1;
	}

	// 设置回调函数
	// 参数：句柄、回调函数
	mosquitto_connect_callback_set(mosq, mqtt_connect_callback);
	mosquitto_disconnect_callback_set(mosq, mqtt_disconnect_callback);
	mosquitto_subscribe_callback_set(mosq, mqtt_subscribe_callback);	//设置订阅回调。当代理响应订阅请求时调用。
	mosquitto_message_callback_set(mosq, mqtt_message_callback);		//设置消息回调。当从代理接收到消息时调用它。
	mosquitto_publish_callback_set(mosq, mqtt_publish_callback);	//设置发布回调。当使用mosquitto_publish启动的消息已成功发送到代理时，将调用此方法。

	// 连接至服务器
	// 参数：句柄、ip（host）、端口、心跳
	char host[100]="",port[20];
	int iport = 1883;

	memset(host,0,sizeof(host));
	memset(port,0,sizeof(port));
	SplitStringKeyValue(psudata->server_mqttserver,host,port);
	if(strlen(port) != 0) iport = StrToInt(port);
	if(iport == 0)iport = 1883;
	psus_hint("mqtt host:[%s:%d]", host,iport);
	// mosq->username = "smartiot";
	// mosq->password = "smartiot";
	mosquitto_username_pw_set(mosq,"smartiot","smartiot");
	ret = mosquitto_connect(mosq, host, iport, KEEP_ALIVE);
	if(ret){
		psus_error("MQTT Connect server error!");
		// mosquitto_destroy(mosq);
		// mosquitto_lib_cleanup();
		// mosq = NULL;
		// mqtt_connected = 0;
		// sgw_set_online(0);
		// return -1;
	}

	ret = mosquitto_loop_start(mosq);
	if(ret){
		psus_hint("MQTT Start loop error!");
		mosquitto_destroy(mosq);
		mosquitto_lib_cleanup();
		mosq = NULL;
		mqtt_connected = 0;
		aexgpio_set_online_status(0);
		return -1;
	}
	// 开始通信：循环执行、直到运行标志g_iRunFlag被改变
	psus_hint("MQTT Start!");
	return 0;
}

int mqtt_quit()
{
	// 结束后的清理工作
	if(mosq){
		mosquitto_loop_stop(mosq, false);
		mosquitto_destroy(mosq);
		mosquitto_lib_cleanup();
		psus_hint("MQTT End!");
	}
	aexgpio_set_online_status(0);
	return 0;
}

#endif
